package producer

import (
	"encoding/json"
	"gin-vue-admin/global"
	"gin-vue-admin/kafka"
	"go.uber.org/zap"
	"log"

	"github.com/Shopify/sarama"
)

type AsyncProducer struct {
	sarama.AsyncProducer
	topic string
}

func newAsyncProducer(topic string) *AsyncProducer {
	pro := kafka.NewAsyncProducer()
	e := &AsyncProducer{
		pro,
		topic,
	}
	go e.asyncDealMessage()
	return e
}

func (e *AsyncProducer) asyncDealMessage() {
	for {
		select {
		case res := <-e.Successes():
			log.Println("push msg success", "topic is", res.Topic, "partition is ", res.Partition, "offset is ", res.Offset)
		case err := <-e.Errors():
			global.GVA_LOG.Error("发送kafka信息失败!", zap.Any("err", err))
		}
	}
}

func (e *AsyncProducer) producer(data *kafka.Msg) {
	bytes, err := json.Marshal(data)
	if err != nil {
		global.GVA_LOG.Error("解析kafka信息失败!", zap.Any("err", err))
		return
	}
	e.Input() <- &sarama.ProducerMessage{
		Value: sarama.ByteEncoder(bytes),
		Topic: e.topic,
	}
}

var asyncProducer *AsyncProducer = newAsyncProducer(kafka.Topic)

func GetAsyncProducer() *AsyncProducer {
	return asyncProducer
}
